package com.kygeng.dataupdate.consumer;

import com.kygeng.common.config.KafkaConfig;
import com.kygeng.dataupdate.function.CompanySearchFunction;
import com.kygeng.common.model.document.CompanyInfoDoc;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;


import com.alibaba.fastjson.JSONObject;

import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.util.CollectionUtils;

import java.util.ArrayList;
import java.util.List;

@Component
@Slf4j
public class KafkaCompanyDataConsumer {

    @Autowired
    private CompanySearchFunction companySearchFunction;

    @KafkaListener(topics = KafkaConfig.TOPIC_COMPANY, groupId = KafkaConfig.GROUP_ID_COMPANY)
    public void topicCompanyConsumer(List<String> messages, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {

        List<String> insertCompanyCodeData = new ArrayList<>();
        List<String> deleteCompanyCodeData = new ArrayList<>();
        List<String> updateCompanyCodeData = new ArrayList<>();

        for (String message : messages) {
            JSONObject messageJSON = JSONObject.parseObject(message);
            String type = messageJSON.getString("type");
            String companyCode = messageJSON.getString("companyCode");
            switch (type) {
                case "insert":
                    insertCompanyCodeData.add(companyCode);
                    break;
                case "update":
                    deleteCompanyCodeData.add(companyCode);
                    break;
                case "delete":
                    updateCompanyCodeData.add(companyCode);
                    break;
                default:
                    break;
            }
        }

        if (!CollectionUtils.isEmpty(insertCompanyCodeData)) {
            insetDataToCompanyEs(insertCompanyCodeData);
        }
        if (!CollectionUtils.isEmpty(deleteCompanyCodeData)) {
            deleteDataFromCompanyEs(deleteCompanyCodeData);
        }
        if (!CollectionUtils.isEmpty(updateCompanyCodeData)) {
            updateDataToCompanyEs(updateCompanyCodeData);
        }
    }

    public void insetDataToCompanyEs(List<String> companyCodes) {
        //TODO: 批量插入数据到 ES
    }

    public void deleteDataFromCompanyEs(List<String> companyCodes) {
        //TODO:删除ES
    }

    public void updateDataToCompanyEs(List<String> companyCodes) {
        //TODO:更新ES
    }

    public List<CompanyInfoDoc> getDataFromCompany(List<String> companyCodes) {
        return companySearchFunction.getCompanyInfoDocBatch(companyCodes);
    }

}
